package com.chief.sink.elasticsearch;


import co.elastic.clients.elasticsearch.ElasticsearchClient;
import co.elastic.clients.elasticsearch.core.*;
import co.elastic.clients.elasticsearch.core.search.Hit;
import co.elastic.clients.elasticsearch.core.search.HitsMetadata;
import co.elastic.clients.json.jackson.JacksonJsonpMapper;
import co.elastic.clients.transport.ElasticsearchTransport;
import co.elastic.clients.transport.rest_client.RestClientTransport;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.source.RichParallelSourceFunction;
import org.apache.http.HttpHost;
import org.elasticsearch.client.RestClient;

import java.util.List;
import java.util.function.Consumer;

public class EsSource extends RichParallelSourceFunction<JSONObject> {
    ElasticsearchClient client =null;
    CreateRequest.Builder<String> builder=null;

    private boolean running = true;

    @Override
    public void open(Configuration parameters) throws Exception {
        RestClient restClient = RestClient.builder(
                new HttpHost("hadoop52", 9200),
                new HttpHost("hadoop53", 9200),
                new HttpHost("hadoop54", 9200),
                new HttpHost("hadoop55", 9200)
        ).build();
        ElasticsearchTransport transport = new RestClientTransport(
                restClient, new JacksonJsonpMapper());
        client = new ElasticsearchClient(transport);
        builder = new CreateRequest.Builder<>();
    }

    @Override
    public void close() throws Exception {
        client.shutdown();
        builder=null;
    }

    @Override
    public void run(SourceContext<JSONObject> ctx) throws Exception {
       if(running){
           SearchRequest build = new SearchRequest.Builder().index("products").build();
           SearchResponse<JSONObject> search = client.search(build, JSONObject.class);
           List<Hit<JSONObject>> hits = search.hits().hits();
           hits.forEach(stringHit -> {
               try {
                   Thread.sleep(10000);
               } catch (InterruptedException e) {
                   e.printStackTrace();
               }
               System.out.println(stringHit);
               ctx.collect( stringHit.source());
           });
       }
    }

    @Override
    public void cancel() {
        running=false;
    }
}
